%%-----------------------------------------------------------------------------
%% Author: vjache
%% Created: 27.02.2009
%% Description: Basic key-value implemetation of vector based on mnesia.
%%-----------------------------------------------------------------------------
-module(mnesia_vect).

%%
%% Include files
%%
-include("../include/rrs.hrl").

-record(vector,{
				name,
				drv_module=?MODULE,
				residence=disk,
				type=bag}).
-record(entry,{key,val}).

%%-define(QALIFIED_KEY(Ns,Key),{Key,Ns}).
%%-----------------------------------------------------------------------------
%% Exported Functions
%%-----------------------------------------------------------------------------
-export([
		 open_vector/2,
		 delete_vector/1,
		 get/3,
		 get/2,
		 fold_slice/4,
		 fold/3,
		 pfold/4,
		 put/3,
		 get_next/1,
		 clear/1,
		 size/1,
		 activity/1]).

-export([
		 test/1]).

%%-----------------------------------------------------------------------------
%% API Functions
%%-----------------------------------------------------------------------------

%% start_link()->ok
%% 
%% create_vector(Name,Flags) ->
%% 	Vector

%%--------------------------------------------------------------------
%% Function: open_vector(Name,Context) -> vector()|
%%										  {error,{no_such_vect,Name}}
%% Description: Opens an existing vector using context. Context may be 
%%				either 'local' or 'global'. If context local, then all
%%				operations addresses to local component, that is to the
%%				partition located on node where caller process resides.
%%				If context global, then all operations use ring route
%%				info (see svc_rrs module) to access vector partitions 
%%				distributed over of cluster.
%%--------------------------------------------------------------------
open_vector(Name,_Context=global) ->
	NRouteSnap=svc_rrs:make_route_snapshot(),
	RouteNodes=route_snap_to_nodes(NRouteSnap),
	RetVal=is_vect_exists(Name),
	case RetVal of
		{error,{no_such_vect,Name}} -> RetVal;
		Vector when is_record(Vector,vector) -> 
			lists:foreach(fun(Node)->
								  ensure_part_exists(Vector,Node)
						  end, RouteNodes),
			{global,Vector,NRouteSnap}
		
	end;
open_vector(Name,_Context=local) ->
	RetVal=is_vect_exists(Name),
	case RetVal of
		false -> {error,{no_such_vect,Name}};
		Vector when is_record(Vector,vector) -> 
			ensure_part_exists(Vector,node())
	end.

delete_vector({global,Vect,NRouteSnap}) ->
	Nodes=route_snap_to_nodes(NRouteSnap),
	Proms=[rpc:async_call(Node, ?MODULE, delete_vector, [{local,Vect,tab_name(Vect,Node)}])||Node<-Nodes],
	RpcRes=lists:foldl(fun(Prom,{Oks,Bads}) ->
						case rpc:yield(Prom) of
							{badrpc,_Reason}=Bad ->
								{Oks,[Bad|Bads]};
							Res ->
								{[Res|Oks],Bads}
						end
				end,{[],[]},Proms),
	case RpcRes of
		{_,[]}->
			ok;
		{_,Bads}->
			{error,Bads}
	end;
delete_vector({local,_Vect,TabName}) ->
	{atomic,ok}=mnesia:delete_table(TabName),
	ok.
	

%%--------------------------------------------------------------------
%% Function: put(Handle,Namespace,KeyFunOrKeyFun,Values) -> ok
%% Description: Puts a set of values under key to the vector accordinly 
%%				context to the specified Namespace . As a key can be passed
%%				an explicit key or 'fun(Value)->{Key,Value1}' that extracts key
%%				from value.
%%--------------------------------------------------------------------
put({local,_Vect,TabName},KeyFun,Values) when is_function(KeyFun)->
	R=activity(fun lists:foreach/2,
			 [fun(Value) ->
					  {K,V}=KeyFun(Value),
					  mnesia:write(TabName,#entry{key=K,val=V},write)
			  end,
			  Values]),
	R;
put({local,_Vect,_TabName}=H,Key,Values) ->
	put(H,
		fun(_)->
				Key
		end,Values);
put({global,Vect,NRouteSnap},Key,Values) when not is_function(Key)->
	Node=?ROUTE2UPNODE(svc_rrs:key_route(Key, NRouteSnap)),
	ok=rpc:call(Node, ?MODULE, put, [{local,Vect,tab_name(Vect,Node)},Key,Values]);
put({global,Vect,NRouteSnap},KeyFun,Values) when is_function(KeyFun)->
	NVDicts=pvalue_route(KeyFun,Values,NRouteSnap),
	io:format("NVDicts length=~p~n", [length(NVDicts)]),
	Proms=lists:foldl(fun(NVDict,AccIn) ->
							  [dict:fold(fun(Node,NValues,Promises)->
												 [rpc:async_call(Node, ?MODULE, put, [{local,Vect,tab_name(Vect,Node)},KeyFun,NValues])|Promises]
										 end
										 , [], NVDict),AccIn]
					  end,[],NVDicts),
	RpcRes=lists:foldl(fun(Prom,{Oks,Bads}) ->
							   case rpc:yield(Prom) of
								   {badrpc,_Reason}=Bad ->
									   {Oks,[Bad|Bads]};
								   Res ->
									   {[Res|Oks],Bads}
							   end
					   end,{[],[]},lists:flatten(Proms)),
	case RpcRes of
		{_,[]}->
			ok;
		{_,Bads}->
			{error,Bads}
	end.

%%--------------------------------------------------------------------
%% Function: get(Handle,Key) -> {Page,Continuation}
%% Description: Gets a set of values under specified key. The result 
%% 				returned as tuple with limited list(page) of values and 
%%				opaque term	that can be used to get next page using 
%%				fun 'get_next/1'. 
%%--------------------------------------------------------------------
get({local,_Vect,TabName},Key) ->
	MatchHead = #entry{key=Key, val='$1'},
    ResultSpec = '$1',%{Key,'$1'},
	Res=activity(
		  fun()->
				  mnesia:select(TabName, 
								[{MatchHead, [], [ResultSpec]}], 
								1000, read)
		  end),
	case Res of
		{_Page,_Cont}=Ret -> Ret;
		'$end_of_table' -> '$EOF'
	end;
get({global,Vect,NRouteSnap},Key) ->
	Node=?ROUTE2UPNODE(svc_rrs:key_route(Key, NRouteSnap)),
	rpc:call(Node, ?MODULE, get, [{local,Vect,tab_name(Vect,Node)},Key]).
get({global,Vect,NRouteSnap},Key,ValueFilterFun) ->
	Node=?ROUTE2UPNODE(svc_rrs:key_route(Key, NRouteSnap)),
	rpc:call(Node, ?MODULE, get, [{local,Vect,tab_name(Vect,Node)},Key,ValueFilterFun]);
get({local,_Vect,_TabName}=H,Key,ValueFilterFun) when is_function(ValueFilterFun)->
	case get(H,Key) of
		{Page,Cont} -> {lists:filter(ValueFilterFun,Page),Cont};
		'$EOF'=Eof -> Eof
	end.
get_next(Cont) ->
	Res=activity(
		  fun()->
				  mnesia:select(Cont)
		  end),
	case Res of
		{_Page,_Cont}=Ret -> Ret;
		'$end_of_table' -> '$EOF'
	end.

%%--------------------------------------------------------------------
%% Function: clear(Handle) -> ok
%% Description: Clears vector. That is removes all data in the vector. 
%%--------------------------------------------------------------------
clear({global,Vect,NRouteSnap}) ->
	Nodes=route_snap_to_nodes(NRouteSnap),
	Proms=[rpc:async_call(Node, ?MODULE, clear, [{local,Vect,tab_name(Vect,Node)}])||Node<-Nodes],
	RpcRes=lists:foldl(fun(Prom,{Oks,Bads}) ->
						case rpc:yield(Prom) of
							{badrpc,_Reason}=Bad ->
								{Oks,[Bad|Bads]};
							Res ->
								{[Res|Oks],Bads}
						end
				end,{[],[]},Proms),
	case RpcRes of
		{_,[]}->
			ok;
		{_,Bads}->
			{error,Bads}
	end;
clear({local,_Vect,TabName}) ->
	FragNames=(catch activity(fun()-> 
									  mnesia:table_info(TabName,frag_names) 
							  end)),
	case FragNames of
		{'EXIT',{aborted,{no_exists,TabName,frag_names}}} ->
			ok;
		{'EXIT',{aborted,{R}}} ->
			exit(R);
		_ ->
			lists:foreach(fun(FragName)->
								  mnesia:clear_table(FragName)
						  end, FragNames)
	end.
%%--------------------------------------------------------------------
%% Function: size(Handle) -> int()
%% Description: Returns a number of entries in the vector. 
%%--------------------------------------------------------------------
size({global,Vect,NRouteSnap}) ->
	Nodes=route_snap_to_nodes(NRouteSnap),
	Proms=[rpc:async_call(Node, ?MODULE, size, [{local,Vect,tab_name(Vect,Node)}])||Node<-Nodes],
	RpcRes=lists:foldl(fun(Prom,{Oks,Bads}) ->
						case rpc:yield(Prom) of
							{badrpc,_Reason}=Bad ->
								{Oks,[Bad|Bads]};
							Res ->
								{[Res|Oks],Bads}
						end
				end,{[],[]},Proms),
	case RpcRes of
		{Sizes,[]}->
			lists:sum(Sizes);
		{_,Bads}->
			{error,Bads}
	end;
size({local,_Vect,TabName}) ->
	activity(fun()->
					case (catch mnesia:table_info(TabName,size)) of
						Size when is_integer(Size) ->
							Size;
						{'EXIT',{aborted,{no_exists,TabName,size}}} ->
							0;
						{'EXIT',{aborted,R}} ->
							exit(R)
					end
			 end).

fold_slice({global,Vect,NRouteSnap},Key,FoldFun,Acc0) ->
	Node=?ROUTE2UPNODE(svc_rrs:key_route(Key, NRouteSnap)),
	rpc:call(Node, ?MODULE, fold_slice, [{local,Vect,tab_name(Vect,Node)},Key,FoldFun,Acc0]);
fold_slice({local,_Vect,_TabName}=H,Key,FoldFun,Acc0) ->
	fold_pages(get(H,Key), FoldFun,Acc0).

fold({global,Vect,NRouteSnap},FoldFun,Acc0) ->
	Nodes=route_snap_to_nodes(NRouteSnap),
	lists:foldl(fun(Node,AccIn)->
						Res=rpc:call(Node, ?MODULE, fold, [{local,Vect,tab_name(Vect,Node)},FoldFun,AccIn]),
						case Res of
							{badrpc,_}=Err ->
								exit({error,Node,Err});
							_ ->
								Res
						end
				end, Acc0, Nodes);
fold({local,_Vect,TabName}=_H,FoldFun,Acc0) ->
	activity(fun()->
					 mnesia:foldl(fun(#entry{key=K,val=V},AccIn)->
										  FoldFun({K,V},AccIn)
								  end, Acc0, TabName)
			 end).

pfold({global,Vect,NRouteSnap},FoldFun,AccMergeFun,Acc0) ->
	Nodes=route_snap_to_nodes(NRouteSnap),
	%io:format("1~n"),
	Proms=[rpc:async_call(Node, ?MODULE, fold, [{local,Vect,tab_name(Vect,Node)},FoldFun,Acc0])
			 ||Node<-Nodes],
	%io:format("~p~n",[Proms]),
	RpcRes=lists:foldl(fun(Prom,{AccIn,Bads}) ->
							   Y=rpc:yield(Prom),
							   %io:format("Y=~p~n",[Y]),
							   case Y of
								   {badrpc,_Reason}=Bad ->
									   {AccIn,[Bad|Bads]};
								   AccRpc ->
									   {AccMergeFun(AccRpc,AccIn),Bads}
							   end
					   end,{Acc0,[]},Proms),
	case RpcRes of
		{AccTot,[]}->
			AccTot;
		{_,Bads}->
			{error,Bads}
	end.
%% remove(Vector,Namespace,Key):ok
%% remove(Vector,Namespace,Key,Filter):ok
%% keys(Vector,Namespace):[Key]
%% keys(Vector,Namespace,Filter):[Key]



%%-----------------------------------------------------------------------------
%% Local Functions
%%-----------------------------------------------------------------------------
route_snap_to_nodes(NRouteSnap) ->
	gb_sets:to_list(
				 lists:foldl(fun(Route,AccIn)->
									 Node=?ROUTE2UPNODE(Route),
									 gb_sets:add(Node, AccIn)
							 end, 
							 gb_sets:new(), 
							 gb_trees:values(NRouteSnap))).

is_vect_exists(VectName) ->
	#vector{name=VectName,residence=ram}.
tab_config(#vector{type=bag,residence=disk,drv_module=?MODULE,name=_Name}=Vec,Node) ->
	{_TabName=tab_name(Vec,Node),
	 _TabDef=[	{record_name, entry},
				{attributes, record_info(fields,entry)},
				{disc_only_copies, [Node]},
				%{index, [#entry.ns]},
				{type, bag},
				{frag_properties, 
					[	{n_fragments, 20},
						{n_disc_only_copies, 1},
						{node_pool, [Node]}
					]
				}
			]
	};
tab_config(#vector{type=bag,residence=ram,drv_module=?MODULE,name=_Name}=Vec,Node) ->
	{_TabName=tab_name(Vec,Node),
	 _TabDef=[	{record_name, entry},
				{attributes, record_info(fields,entry)},
				{ram_copies, [Node]},
				%{index, [#entry.ns]},
				{type, bag},
				{frag_properties, 
					[	{n_fragments, 20},
						{n_ram_copies, 1},
						{node_pool, [Node]}
					]
				}
			]
	}.
tab_name(Vec,Node) ->
	list_to_atom(
				atom_to_list(Vec#vector.name)++[$.|atom_to_list(Node)]).

ensure_part_exists(Vector,Node) ->
	{TabName,TabConf}=tab_config(Vector,Node),
	%% 1. Check Vector registered in CRS
	%% 2. Read info for Vector
	%% 3. Check local partition exists
	case (catch mnesia:table_info(TabName,record_name)) of
		entry -> ok;
		{'EXIT',{aborted,{no_exists,TabName,record_name}}} -> 
			create_part(TabName,TabConf);
		{'EXIT',{aborted,Reason}} -> exit(Reason)
	end,
	{local,Vector,TabName}.

create_part(TabName,TabDef) ->
	Result=mnesia:create_table(TabName, TabDef),
	{atomic,ok}=Result.

activity(Fun) ->
	activity(Fun,[]).
activity(Fun, Args) ->
	mnesia:activity(async_dirty, Fun, Args, mnesia_frag).

fold_pages('$EOF',_FoldFun,Acc0) ->
	Acc0;
fold_pages({Page,Cont},FoldFun,Acc0) ->
	AccOut=lists:foldl(FoldFun, Acc0, Page),
	fold_pages(get_next(Cont),FoldFun,AccOut).

value_route(KeyFun,FoldFun,Acc0,Values,NRouteSnap) ->
	lists:foldl(fun(V,AccIn) ->
						{K,_V1}=KeyFun(V),
						Node=?ROUTE2UPNODE(svc_rrs:key_route(K, NRouteSnap)),
						FoldFun(Node,V,AccIn)
				end, Acc0, Values).

value_route(KeyFun,Values,NRouteSnap) ->
	value_route(KeyFun,fun(Node,Value,Dict0) ->
							   dict:append(Node, Value, Dict0)
				end,dict:new(),Values,NRouteSnap).
pvalue_route(KeyFun,Values,NRouteSnap) ->
	R=pvalue_route(KeyFun,
				 fun(Node,Value,Dict0) ->
						 dict:append(Node, Value, Dict0)
				 end,
				 fun(A1,A2)->
						 [A1,A2]
				 end,dict:new(),Values,NRouteSnap),
	lists:flatten([R]).
pvalue_route(KeyFun,FoldFun,AccMergeFun,Acc0,Values,NRouteSnap) ->
	ValListSize=length(Values),
	ChunkMaxSize=round(ValListSize/erlang:system_info(schedulers)),
	ValChunks=split_list(lists:max([1000,ChunkMaxSize]),ValListSize,Values,[]),
	Self=self(),
	Workers=[spawn_link(fun()->
						AccRet=value_route(KeyFun,FoldFun,Acc0,ValChunk,NRouteSnap),
						Self!{self(),AccRet}
				end)||ValChunk<-ValChunks],
	rec_loop(length(Workers),AccMergeFun,Acc0).

rec_loop(0,_AccMergeFun,Acc0) ->
	Acc0;
rec_loop(WorkersNum,AccMergeFun,Acc0) ->
	receive
		{_Pid,Acc}->
			rec_loop(WorkersNum-1,AccMergeFun,AccMergeFun(Acc,Acc0))
	end.

split_list(ChunkMaxSize,ListSize,List,Result) when ListSize=<ChunkMaxSize ->
	[List|Result];
split_list(ChunkMaxSize,ListSize,List,Result) when ListSize>ChunkMaxSize ->
	{Chunk,Rest}=lists:split(ChunkMaxSize, List),
	split_list(ChunkMaxSize,ListSize-ChunkMaxSize,Rest,[Chunk|Result]).


%%------------------------------- Tests ---------------------------------------
data(Iters) ->
	{A,B,C}=now(),
	random:seed(A,B,C),
	Range=10000000,
	[{{"some arbitrary key",random:uniform(Range)},{"some arbitrary value",random:uniform(Range)}}||_N<-lists:seq(1,Iters)].
test({put,Ctx,DataSize}) ->
	H=open_vector(kv, Ctx),
	lists:foreach(fun({K,V})->
						  put(H,K,[V])
				  end,data(DataSize)),
	H;
test({put_bp,Ctx,DataSize}) ->
	H=open_vector(kv, Ctx),
	put(H,fun({K,V})->
				  {K,V}
		end,data(DataSize)),
	H;
test({put,Ctx}) ->
	test({put,Ctx,3000});
test(pfold) ->
	H=open_vector(kv, global),
	D=pfold(H, fun({K,_V},Acc)->
					 dict:update_counter(K, 1, Acc)
		  end, 
		  fun(A1,A2)->
				  dict:merge(fun(_K,V1,V2)->
									 V1+V2
							 end, A1, A2)
		  end, dict:new()),
	dict:to_list(D).
tt()->
	lists:foldl(fun(T,AccIn)->basic_vect:activity(fun()-> mnesia:table_info(T,size) end)+AccIn end,0,['kv.demetra0@ws24292','kv.demetra1@ws24292','kv.demetra2@ws24292','kv.demetra3@ws24292']).
%%-----------------------------------------------------------------------------